package cn.itcast.type;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

// 消费者
// 接口可以new 但是要实现它的方法
// 先完成消费者的配置和消息监听器的注册，最后启动消费者，这样能确保消费者在启动时已做好接收和处理消息的准备
public class Consumer {
    public static void main(String[] args) throws Exception {
        // 1. 谁来收?
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        // 2. 从哪里收消息?
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 3. 订阅什么? 监听那个消息队列
        consumer.subscribe("topic1", "*");

        // 4. 处理业务流程 注册监听器 推送消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 写我们的业务逻辑
                list.forEach(messageExt -> {
                    System.out.println(messageExt);
                    byte[] body = messageExt.getBody();
                    System.out.println(new String(body));
                });
                // 业务逻辑处理完成
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 5. 开始收？
        consumer.start();
        // 6. 千万不要关闭消费者
    }
}
